package com.hadoop.multitable;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class MultiTableJoin {

	public static int time = 0;

	public static class Map extends Mapper<LongWritable, Text, Text, Text> {

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String[] record = line.split(" ");
			if (record[0].compareTo("factoryName") != 0 && record[0].compareTo("addressId") != 0) {
				int point = line.lastIndexOf(" ");
				String firstPart = line.substring(0, point);
				String lastPart = line.substring(point + 1);
				if (this.isNumeric(firstPart)) {
					context.write(new Text(firstPart), new Text("address#" + lastPart));
				} else {
					context.write(new Text(lastPart), new Text("factory#" + firstPart));
				}
			} // if
		}

		private boolean isNumeric(String value) {
			Pattern pattern = Pattern.compile("[0-9]*");
			Matcher isNum = pattern.matcher(value);
			if (isNum.matches()) {
				return true;
			}
			return false;
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Text factoryName = new Text("factoryName");
			Text factoryAddress = new Text("factoryAddress");

			if (time == 0) {
				context.write(factoryName, factoryAddress);
				time++;
			}

			List<Text> factoryList=new ArrayList<Text>();
			for (Text value : values) {// pair: name address
				String[] flagAndInfo = value.toString().split("#");
				if (flagAndInfo[0].compareTo("factory") == 0) {
					factoryList.add(new Text(flagAndInfo[1]));
				} else {
					factoryAddress.set(flagAndInfo[1]);
				}
			}
			
			for(Text factory:factoryList){
				context.write(factory, factoryAddress);
			}
			
		}// reduce
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// TODO Auto-generated method stub

		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: MultiTableJoin <in> <out>");
			System.exit(2);
		}

		Job job = Job.getInstance(conf, "Multi Table Join");
		job.setJarByClass(MultiTableJoin.class);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}
